package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.Date;
import java.util.UUID;

/**
 * @Author 派大星
 * @Description TODO
 * @Date 2025/7/30 16:03
 */
public class IRetryTest {

    @Test
    public void retryProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("retry-producer-group");
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        producer.start();
        // 生产者发送消息重试次数
        producer.setRetryTimesWhenSendFailed(2);
        producer.setRetryTimesWhenSendAsyncFailed(2);

        String key = UUID.randomUUID().toString();
        System.out.println(key);
        // 创建一个带tag的消息
        Message message1 = new Message("retryTopic", "vip1", key, "我是消息666".getBytes());
        producer.send(message1);
        System.out.println("发送成功");
        producer.shutdown();
    }

    @Test
    public void retryConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("retryTopic", "*");
        // 设置重试次数
        consumer.setMaxReconsumeTimes(3);
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExtList, context) -> {
            MessageExt messageExt = messageExtList.get(0);
            System.out.println(new Date());
            System.out.println(messageExt.getReconsumeTimes());
            System.out.println(new String(messageExt.getBody()));
            try {
                handleDb();
            } catch (Exception e) {
                if (messageExt.getReconsumeTimes() >= 3) {
                    // 不要再消费了
                    System.out.println("记录到特别的位置 文件 mysql 通知人工处理");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            // 业务报错了 返回null 返回RECONSUME_LATER 都会进行重试
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }

    private void handleDb() {
        int i = 10 / 0;
    }

    @Test
    public void retryDeadConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-dead-consumer-group");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("%DLQ%retry-consumer-group", "*");
        // 死信主题中的数据就交给人工来进行处理 主要是发邮件提醒之类的
        consumer.setMaxReconsumeTimes(2);
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExtList, context) -> {
            MessageExt messageExt = messageExtList.get(0);
            System.out.println(new Date());
            System.out.println(messageExt.getReconsumeTimes());
            System.out.println(new String(messageExt.getBody()));
            // 业务报错了 返回null 返回RECONSUME_LATER 都会进行重试
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}
